package offerset;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.IntegerSerializer;
import org.apache.kafka.common.serialization.StringSerializer;

import java.time.Duration;
import java.util.Arrays;
import java.util.Properties;

/**
 * @Author:RenPu
 * @Date: 2020/3/18 19:33
 * @Version: 1.0
 * @description:偏移量位置的控制，设置禁止自动提交，手动提交。
 */
public class KafkaConsumerDemo {

    public static void main(String[] args) {

        //创建配置对象，指定Consumer的配置参数
        Properties properties = new Properties();

        //配置消费者连接kafaka集群的参数
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"");

        //配置recoder的反序列化的参数配置
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, IntegerSerializer.class);

        //配置recoder的反序列化的参数配置
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringSerializer.class);

        //配置消费组的ID值，组内负载均衡，组外广播
        properties.put(ConsumerConfig.GROUP_ID_CONFIG,"g1");

        //关闭自动提交offerset偏移量
        properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,false);


        //创建消费者对象
        KafkaConsumer<Integer, String> consumer = new KafkaConsumer<Integer, String>(properties);

        //订阅主题
        consumer.subscribe(Arrays.asList("t1"));


        //拉取主题内新增数据内容
        while(true){

            //拉取主题内的最新数据，并设置拉取的超时时间
            ConsumerRecords<Integer, String>  records= consumer.poll(Duration.ofSeconds(5));

            for (ConsumerRecord<Integer, String> record : records) {

                System.out.println(record.key()+"|"+record.value()+"|"+record.timestamp()+"|"+record.partition()+"|"+record.offset());

            }

            //手动提交偏移量(异步提交)
            consumer.commitAsync();

        }




    }


}
